package com.ly.rabbitmq.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @Classname RabbitMQConfig
 * @Description TODO
 * @Date 2020/11/20 20:07
 * @Author 冷心影翼
 */
//@Configuration
public class RabbitMQConfig {

	public static final String EXCHANGE_NAME = "bootExchangeTopic";

	public static final String QUEUE_NAME = "bootQueue";

	@Autowired
	RabbitTemplate rabbitTemplate;

	@Autowired
	private CachingConnectionFactory connectionFactory;

	@Bean
	public MessageConverter messageConverter() {
		return new Jackson2JsonMessageConverter();
	}

	@PostConstruct
	public void initRabbitTemplate() {
		/**
		 * @param correlationData 配置信息
		 * @param ack  交换机是否成功收到消息  true 成功
		 * @param cause 失败的原因
		 */
		rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
			System.out.println("corre: "+correlationData+" ack:"+ack+" cause: "+cause);
		});

		/**
		 * 只有消息没有投递给指定的队列，就会触发这个方法
		 * @param message 失败投递的信息
		 * @param replyCode 回复的状态码
		 * @param replyText 回复的文本内容
		 * @param exchange  这个消息指定的交换机
		 * @param routingkey 这个消息指定的路由件
		 */
		rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingkey) -> {
			System.out.println(message+" "+replyCode+" "+replyText+" "+exchange+" "+routingkey);
		});
	}



	@Bean("bootExchangeTopic")
	public Exchange bootExchangeTopic() {
		return ExchangeBuilder.topicExchange(EXCHANGE_NAME)
			.durable(true)
			.build();
	}

	@Bean("bootQueue")
	public Queue bootQueue() {
		return QueueBuilder.durable(QUEUE_NAME).build();
	}

	@Bean
	public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue,@Qualifier("bootExchangeTopic") Exchange ex) {
		return BindingBuilder.bind(queue).to(ex).with("#.error").noargs();
	}


	@Bean("ttlQueue")
	public Queue ttlQueue() {
		Map<String,Object> map = new HashMap<>();
		map.put("x-dead-letter-exchange", EXCHANGE_NAME);
		map.put("x-dead-letter-routing-key", "dead.ttl");
		map.put("x-message-ttl", 5000);
		return new Queue("ttlQueue", true, false, false,map);
	}


	@Bean("deadQueue")
	public Queue ttlDeadQueue() {
		Map<String,Object> map = new HashMap<>();
		map.put("x-message-ttl", 10000);
		return new Queue("deadQueue", true, false, false,map);
	}

	@Bean
	public Binding bindDeadQueueToExchange(@Qualifier("deadQueue") Queue queue,@Qualifier("bootExchangeTopic") Exchange ex) {
		return BindingBuilder.bind(queue).to(ex).with("dead.#").noargs();
	}

	@Bean
	public Binding bindTtlQueueToExchange(@Qualifier("ttlQueue") Queue queue,@Qualifier("bootExchangeTopic") Exchange ex) {
		return BindingBuilder.bind(queue).to(ex).with("ttl.#").noargs();
	}


	@Bean("bootContainerFactory")
	public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory() {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		//最大连接数
		factory.setConcurrentConsumers(5);
		//限流
		factory.setPrefetchCount(10);
		factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
		factory.setMessageConverter(new Jackson2JsonMessageConverter());
		return factory;
	}
}
